package com.yangcangclub.mq.domain;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.yangcangclub.mq.base.MQBaseDomain;
import com.yangcangclub.mq.bean.MQTopicInfo;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @author flyger
 * @date 2016-07-21
 * @desc MQTopic domain
 */
@Component
public class MQTopicProfile extends MQBaseDomain {

    /**
     * 添加
     * @param partition
     * @param topic
     * @param server
     * @return
     */
    public  int addMQTopic(int partition,String topic, String server){

        MQTopicInfo topicInfo = new MQTopicInfo();
        topicInfo.setPartition_(partition);
        topicInfo.setServer(server);
        topicInfo.setTopic(topic);

        // 处理并发
        boolean flag=false;
        int expireSeconds=1;
        BoundValueOperations bvo = redisTemplate.boundValueOps("MQTopic.partition." + partition +".topic."+topic+".server."+server+".create.MQTopic.lock");

//        //防止重复
        if(bvo.setIfAbsent(System.currentTimeMillis())) {
            flag=true;
        }
        int id = 0;
        if(flag) {
            super.getMasterJdbc().insert("mq.MQMapper.insertMQTopic",topicInfo);
            id = topicInfo.getId();
//            bvo.expire(expireSeconds, TimeUnit.SECONDS);
            redisTemplate.delete("MQTopic.partition." + partition +".topic."+topic+".server."+server+".create.MQTopic.lock");
        }

//        if(id == 0) {
//            throw new ApolloException(MessageEnum.DATABASE_ERROR);
//        }

        return id;
    }

    /**
     * 获取服务器所有topic partition
     * @param server
     * @return
     */
    public List<MQTopicInfo> getServerMQTopicList(String server){

        Map<String,Object> map = Maps.newHashMap();
        map.put("server",server);

        List<MQTopicInfo> mqTopicInfoList = super.getSlaveJdbc().selectList("mq.MQMapper.selectServerMQTopicList",map);

        if(null==mqTopicInfoList) {
            return Lists.newArrayList();
        }

        return mqTopicInfoList;
    }

    /**
     * 获取服务器具体topic partition
     * @param server
     * @return
     */
    public MQTopicInfo getServerMQTopic(String server,String topic) {

        Map<String,Object> map = Maps.newHashMap();
        map.put("server",server);
        map.put("topic",topic);

        int count = super.getSlaveJdbc().selectOne("mq.MQMapper.selectServerMQTopicCount", map);
        if(0==count) {
            Object maxPartitionObj =  super.getSlaveJdbc().selectOne("mq.MQMapper.selectServerMQTopicMaxPartition", map);
            int maxPartition = 0;
            if(null!=maxPartitionObj) {
                maxPartition = (int)maxPartitionObj;
                maxPartition = maxPartition+1;
            }
            addMQTopic(maxPartition,topic,server);
        }

        MQTopicInfo mqTopicInfo = super.getSlaveJdbc().selectOne("mq.MQMapper.selectServerMQTopic", map);

        return mqTopicInfo;
    }


}
